package com.horizon.im.imServer.distributed;

import org.apache.curator.framework.CuratorFramework;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import com.horizon.constants.ServerConstants;
import com.horizon.entity.ImNode;
import com.horizon.util.JsonUtil;
import com.horizon.zk.CuratorZKClient;

/**
 * 分布式命名服务类： 为im的server端的各个节点生成全局唯一的命名 命名服务的主要实现逻辑是： 所有的工作节点都在ZooKeeper的同一个父节点下创建顺序节点， 然后从返回的临时路径上取得属于自己的后缀编号。
 * 只要有一个netty服务器接入上来，就会初始化一个imworker并在zk的根节点下面创建一个ephemeral_sequential类型节点 获取该节点的后缀作为该netty服务器的id
 */
public class ImWorker {

    private CuratorFramework client = null;

    private String pathRegistered = null;

    private ImNode localNode = null;

    private volatile static ImWorker singletonInstance = null;

    // 获取单例
    public static ImWorker getInstance() {
        if (null == singletonInstance) {
            synchronized (ImWorker.class) {
                if (null == singletonInstance) {
                    singletonInstance = new ImWorker();
                    singletonInstance.client = CuratorZKClient.getInstance().getClient();
                    singletonInstance.localNode = new ImNode();
                }
            }
        }

        return singletonInstance;
    }

    // 1.在zk中创建临时节点
    public void init() {
        createParentIfNeeded(ServerConstants.MANAGE_PATH);

        try {
            // 創建一個inode節點
            // 節點的payload為當前worker實例
            byte[] payload = JsonUtil.object2JsonBytes(localNode);
            pathRegistered = client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
                .forPath(ServerConstants.PATH_PREFIX, payload);
            localNode.setId(getId());
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    private void createParentIfNeeded(String managerPath) {
        try {
            Stat stat = client.checkExists().forPath(managerPath);
            if (null == stat) {
                client.create().creatingParentsIfNeeded().withProtection().withMode(CreateMode.PERSISTENT)
                    .forPath(managerPath);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public ImNode getLocalNode() {
        return localNode;
    }

    public long getId() {
        return getIdByPath(pathRegistered);
    }

    public long getIdByPath(String path) {
        String sid = null;
        if (null == path) {
            throw new RuntimeException("节点路径有误");
        }
        int index = path.lastIndexOf(ServerConstants.PATH_PREFIX);
        if (index >= 0) {
            index += ServerConstants.PATH_PREFIX.length();
            sid = index <= path.length() ? path.substring(index) : null;
        }

        if (null == sid) {
            throw new RuntimeException("节点id获取失败");
        }

        return Long.parseLong(sid);
    }

    // 增加负载，表示有用户登录成功
    public boolean incBalance() {
        if (null == localNode) {
            throw new RuntimeException("还没有设置Node节点");
        }

        // 增加负载，增加负载并写回zk
        try {
            localNode.getBalance().incrementAndGet();
            byte[] payload = JsonUtil.object2JsonBytes(localNode);
            client.setData().forPath(pathRegistered, payload);
            return true;
        } catch (Exception e) {
            return false;
        }
    }

    public boolean decrBalance() {
        if (null == localNode) {
            throw new RuntimeException("还没有设置Node 节点");
        }
        while (true) {
            try {

                localNode.getBalance().decrementAndGet();

                byte[] payload = JsonUtil.object2JsonBytes(localNode);
                client.setData().forPath(pathRegistered, payload);
                return true;
            } catch (Exception e) {
                return false;
            }
        }
    }
}
